package com.shujia.flink.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo02KafkaProducer {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> stuDS = env.readTextFile("flink/data/students.txt");

        KafkaSink<String> kafkaSink = KafkaSink
                .<String>builder()
                .setBootstrapServers("master:9092,node1:9092,node2:9092")
                // 指定数据序列化的方式
                .setRecordSerializer(
                        KafkaRecordSerializationSchema
                                .builder()
                                .setTopic("student01") // 指定Topic
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                // 设置写入数据的语义
                .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();

        stuDS.sinkTo(kafkaSink);

        env.execute();

    }
}
